// -*-mode:c++; coding:utf-8-*-

#include <string>
#include <sstream>

#include <logger.hpp>
#include <performance_measurement.hpp>

#include "group_scanner.hpp"

namespace xbase
{
	static inline
	const char * inner_type_name(group_scanner::record_state::inner_type type) {
		const char *name = "UNKNOWN";
		switch(type)
		{
		case group_scanner::record_state::IT_NO:
			name = "IT_NO";
			break;
		case group_scanner::record_state::IT_ROW:
			name = "IT_ROW";
			break;
		case group_scanner::record_state::IT_FAMILY:
			name = "IT_FAMILY";
			break;
		case group_scanner::record_state::IT_QUALIFIER:
			name = "IT_QUALIFIER";
			break;
		default:
			break;
		}
		return name;
	}

	std::string group_scanner::record_state::to_string() const {
		std::ostringstream out;
		out.setf(ios_base::boolalpha);
		out << "{filter_deleted_record_enabled: " << m_filter_deleted_record_enabled
		    << "; filter_deletion_record_enabled: " << m_filter_deletion_record_enabled
		    << "; is_newest_deletion_record: " << m_is_newest_deletion_record
		    << "; deletion_type: " << inner_type_name(m_deletion_type)
		    << "; filter_version_enabled: " << m_filter_version_enabled
		    << "; version_count: " << m_version_count
		    << "; need_save_status: " << m_need_save_status
		    << "; prev_row: " << dump_string_char(m_prev_row)
		    << "; prev_family: " << m_prev_family
		    << "; prev_qualifier: " << dump_string_char(m_prev_qualifier)
		    << "; prev_timestamp: " << m_prev_timestamp
		    << "; change_type: " << inner_type_name(m_change_type)
		    << "; is_duplicate_record: " << m_is_duplicate_record
		    << "; duplicate_record_deleted: " << m_duplicate_record_deleted
		    << "}";
		return out.str();
	}

	void group_scanner::record_state::update(const block_entry &entry,
						 const column_name_parser &parser) {
		MEASUREMENT_OF_TIME("group_scanner::record_state::update", 0);

		const key_wrapper::string_type &row = *entry.m_key->row();
		//const key_wrapper::string_type &column = *entry.m_key->column();
		const uint64_t timestamp = entry.m_key->timestamp()->get();

		if(m_need_save_status)
		{
			// 判断行列是否变化及其类型，并
			// 更新相应的域；行与列的变化是
			// 独立的，所以分开判断。
			//const column_regex cr(column);
			if(m_prev_row != row) // row changed, then anything changed
			{
				m_change_type = IT_ROW;
				m_prev_row = row;
				m_prev_family = parser.family_name();
				m_prev_qualifier = parser.qualifier_name();
				m_prev_timestamp = timestamp;
				m_is_duplicate_record = false;
				m_duplicate_record_deleted = false;
			}
			else
			{
				if(parser.family_name_cmp(m_prev_family) != 0) // family changed, the whole column changed
				{
					m_change_type = IT_FAMILY;
					m_prev_family = parser.family_name();
					m_prev_qualifier = parser.qualifier_name();
					m_prev_timestamp = timestamp;
					m_is_duplicate_record = false;
					m_duplicate_record_deleted = false;
				}
				else if(parser.qualifier_name_cmp(m_prev_qualifier) != 0) // qualifier changed
				{
					m_change_type = IT_QUALIFIER;
					m_prev_qualifier = parser.qualifier_name();
					m_prev_timestamp = timestamp;
					m_is_duplicate_record = false;
					m_duplicate_record_deleted = false;
				}
				else // not changed
				{
					m_change_type = IT_NO;
					m_is_duplicate_record = (timestamp == m_prev_timestamp);
					if(! m_is_duplicate_record)
					{
						m_duplicate_record_deleted = false;
					}
					m_prev_timestamp = timestamp;
				}
			}
		}

		if(m_filter_deleted_record_enabled)
		{
			// 以前有删除标记，则根据行列的变化情况，更新该删除标记
			if(m_deletion_type != IT_NO)
			{
				switch(m_change_type)
				{
				case IT_NO: // not changed
					break;
				case IT_ROW: // 删除不能跨行
					m_deletion_type = IT_NO;
					break;
				case IT_FAMILY:
					if(m_deletion_type != IT_ROW)
					{
						m_deletion_type = IT_NO;
					}
					// else, 对行的删除仍有效
					break;
				case IT_QUALIFIER:
					if(m_deletion_type == IT_QUALIFIER)
					{
						m_deletion_type = IT_NO;
					}
					// else, 对行和列族的删除仍有效
					break;
				}
			}

			if(m_deletion_type == IT_NO) // 没有遗留的删除标记，则判断这一次的删除标记
			{
				m_is_newest_deletion_record = entry.m_value->empty();
				if(m_is_newest_deletion_record)
				{
					// 去掉对删除行和列族的支持！因此只剩下对小列的删除功能。
					
					// if(m_prev_family.empty())
					// {
					// 	m_deletion_type = IT_ROW;
					// }
					// else if(m_prev_qualifier.empty())
					// {
					// 	m_deletion_type = IT_FAMILY;
					// }
					// else
					// {
					// 	m_deletion_type = IT_QUALIFIER;
					// }
					m_deletion_type = IT_QUALIFIER;
				}
			}
			else
			{
				m_is_newest_deletion_record = false;
			}
		}
				
		if(m_filter_version_enabled)
		{
			if(m_change_type == IT_NO)
			{
				if(! m_is_duplicate_record)
				{
					++ m_version_count;
				}
				// 重复的记录不计入版本数中
			}
			else
			{
				m_version_count = 0;
			}
		}
	}

	void group_scanner::fetch_entry() {
		MEASUREMENT_OF_TIME("group_scanner::fetch_entry", this->id());
		
		for(; ! m_queue.empty(); pop_first())
		{
			// 准备好检查第一个是不是包含到结果中的，若是，则返回；若不是，则去掉，再检查下一个
			const block_entry &entry = m_queue.top()->get();

			// 注意：时间戳对删除记录的影响
			//
			// 下面的实现是，删除记录会无条件删除以前的记
			// 录，即使指定的时间戳范围内不包含它，它也会起
			// 作用。
			//
			// 若想删除记录只在用户指定的时间戳范围内起作
			// 用，则将下面的时间戳判断条件分开并提前，不满
			// 足时直接跳过即可。

			const column_name_parser parser(*entry.m_key->column());
			if((! scan_context_ptr()->is_contain_timestamp(entry.m_key->timestamp()->get()))
			   || (! scan_context_ptr()->is_contain_column(parser))) // 不满足条件
			{
				// 即使列或时间戳不满足，但对第一个删除
				// 记录在以下情况下要特殊处理：过滤被删
				// 除记录设置且过滤删除记录自己关闭，则
				// 需要无条件返回该删除记录。
				//
				// 对非第一个删除记录，不做特殊处理
				if(entry.m_value->empty() // 是删除记录，则可能是第一个删除记录
				   && m_state.m_filter_deleted_record_enabled
				   && (! m_state.m_filter_deletion_record_enabled)) // 条件满足
				{
					m_state.update(entry, parser); // 更新状态
					if(m_state.m_is_duplicate_record || // 是重复记录或
					   (! m_state.m_is_newest_deletion_record)) // 不是第一个删除记录，则可以跳过了
					{
						LOG(DETAIL, "group_scanner::next deletion key ("
						    << dump_string_char(*entry.m_key->row())
						    << ", "
						    << dump_string_char(*entry.m_key->column())
						    << ", "
						    << entry.m_key->timestamp()->get()
						    << ") is out of scan range, and not first deletion record, skip it,"
						    " stat = " << m_state.to_string());
						continue;
					}
					else
					{
						LOG(DETAIL, "group_scanner::next got first deletion record out of scan range,"
						    " stat = " << m_state.to_string());
						// 再经过下面的过滤，因
						// 为是第一条删除记录，
						// 所以会跳过删除检查和
						// 版本检查
					}
				}
				else
				{
					LOG(DETAIL, "group_scanner::next key ("
					    << dump_string_char(*entry.m_key->row())
					    << ", "
					    << dump_string_char(*entry.m_key->column())
					    << ", "
					    << entry.m_key->timestamp()->get()
					    << ") with data size "
					    << entry.m_value->size()
					    << ") is out of scan range, skip it");
					continue;
				}
			}
			else	// 满足条件，则按正常进行，更新扫描状态
			{
				m_state.update(entry, parser);
			}

			// 如果过滤删除记录自己被关闭了，则最新
			// 的删除记录需要跳过版本检查；也就是它
			// 必须被返回（即使版本数已经足够）。但
			// 若版本数还不够，它会占用一个版本（其
			// 实占用和不占用已经没有区别，因为后面
			// 的版本已经被它删除了，不会再返回了）。
			bool is_newest_deletion_record = false;
			if(m_state.m_filter_deleted_record_enabled) // 需要过滤被删除记录
			{
				if(m_state.m_deletion_type != record_state::IT_NO) // 该记录被删除了
				{
					if(m_state.m_is_duplicate_record)
					{
						// 无条件跳过，不保留其
						// 删除功能（仅保留对重
						// 复记录的删除功能，即
						// 设置
						// m_duplicate_record_deleted
						// 为真），使其后面的重
						// 复记录不再返回；这里
						// 不用管它是不是第一条
						// 删除记录和需不需要保
						// 留删除记录本身
						LOG(DETAIL, "deletion record is duplicated, skip it;"
						    << " key = {"
						    << dump_string_char(*entry.m_key->row())
						    << ", " << dump_string_char(*entry.m_key->column())
						    << ", " << entry.m_key->timestamp()->get()
						    << "}, state = " << m_state.to_string());
						m_state.m_deletion_type = record_state::IT_NO;
						m_state.m_duplicate_record_deleted = true;
						continue;
					}
					else if(m_state.m_filter_deletion_record_enabled
						|| (! m_state.m_is_newest_deletion_record))
					{
						LOG(DETAIL, "group_scanner::next key ("
						    << dump_string_char(*entry.m_key->row())
						    << ", "
						    << dump_string_char(*entry.m_key->column())
						    << ", "
						    << entry.m_key->timestamp()->get()
						    << ") filtered by deleted record"
						    ", stat = " << m_state.to_string());
						continue;
					}
					else
					{
						LOG(DETAIL, "group_scanner::next got newset deletion record,"
						    " stat = " << m_state.to_string());
						is_newest_deletion_record = true;
					}
				}
			}

			// 每一条记录的第一个删除记录必须保留！
			// 即永久性删除。为保证行为的一致性，必
			// 须这样实现！否则，后面写入的老版本的
			// 记录（指定时间戳）就可能会被删除，也
			// 可能会保留。
			if(m_state.m_filter_version_enabled) // 需要过滤版本数
			{
				if(m_state.m_is_duplicate_record)
				{
					LOG(DETAIL, "group_scanner::next skip duplicate record, key ("
					    << dump_string_char(*entry.m_key->row())
					    << ", "
					    << dump_string_char(*entry.m_key->column())
					    << ", "
					    << entry.m_key->timestamp()->get()
					    << ") with data size "
					    << entry.m_value->size()
					    << "), stat = " << m_state.to_string());
					continue;
				}
				else
				{
					// 如果过滤删除记录但不过滤最新
					// 删除记录的话，则一定返回最新
					// 的一条删除记录，并无视版本数
					// 的限制，且对版本数没有任何影
					// 响
					const bool skip_version_check = is_newest_deletion_record;
					if(! skip_version_check)
					{
						if(m_state.m_version_count >= scan_context_ptr()->num_version())
						{
							LOG(DETAIL, "group_scanner::next key ("
							    << dump_string_char(*entry.m_key->row())
							    << ", "
							    << dump_string_char(*entry.m_key->column())
							    << ", "
							    << entry.m_key->timestamp()->get()
							    << ") filtered by version num "
							    << scan_context_ptr()->num_version()
							    << ", stat = " << m_state.to_string());
							continue;
						}
						// else, version check ok
					}
					else
					{
						LOG(DETAIL, "group_scanner::next the newest deletion record skips version checking"
						    ", stat = " << m_state.to_string());
					}
				}
			}
			else if(m_state.m_is_duplicate_record &&
				m_state.m_duplicate_record_deleted)
			{
				// 不过滤版本的情况下，若重复记录中曾经
				// 有过删除记录，则后面的重复记录就被认
				// 为是已被删除了。
				LOG(DETAIL, "duplicate record was deleted previously, key ("
				    << dump_string_char(*entry.m_key->row())
				    << ", "
				    << dump_string_char(*entry.m_key->column())
				    << ", "
				    << entry.m_key->timestamp()->get()
				    << ") with data size "
				    << entry.m_value->size()
				    << ", stat = " << m_state.to_string());
				continue;
			}

			// 所有过滤都通过了！
			LOG(DETAIL, "group_scanner::next got key ("
			    << dump_string_char(*entry.m_key->row())
			    << ", "
			    << dump_string_char(*entry.m_key->column())
			    << ", "
			    << entry.m_key->timestamp()->get()
			    << ") with data size "
			    << entry.m_value->size()
			    << ", stat = " << m_state.to_string());
				
			break; // 退出
		}
	}


} // namespace xbase
